package com.group2.edu.realtime.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.group2.edu.realtime.common.base.BaseApp;
import com.group2.edu.realtime.common.bean.DwsExaminationPaperExamWindowBean;
import com.group2.edu.realtime.common.function.DimAsyncFunction;
import com.group2.edu.realtime.common.util.DateFormatUtil;
import com.group2.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 *
 * 需要启动的进程：zk、kafka、flume、doris、DwdBaseLog、DwdLearningPlay，hbase,redis
 */
public class DwsExaminationPaperExamWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsExaminationPaperExamWindow().start(
                10034,
                4,
                "dws_examination_paper_exam_window",
                "dwd_examination_test_paper"

        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> KafkaSource) {
        //KafkaSource.print();
        SingleOutputStreamOperator<DwsExaminationPaperExamWindowBean> dwsflatmap = KafkaSource.flatMap(
                new FlatMapFunction<String, DwsExaminationPaperExamWindowBean>() {
                    @Override
                    public void flatMap(String value, Collector<DwsExaminationPaperExamWindowBean> out) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        long ts = jsonObject.getLong("ts") * 1000;
                        Double score = jsonObject.getDouble("score");

                        out.collect(DwsExaminationPaperExamWindowBean.builder()
                                .paperId(jsonObject.getString("paper_id"))
                                .examTakenCount(1L)
                                .examTotalScore((long) score.doubleValue())
                                .examTotalDuringSec(jsonObject.getLong("duration_sec"))
                                .ts(ts)
                                .build());
                    }

                }
        );
        //dwsflatmap.print();
        SingleOutputStreamOperator<DwsExaminationPaperExamWindowBean> withWaterMarkStream = dwsflatmap.assignTimestampsAndWatermarks(
                WatermarkStrategy.<DwsExaminationPaperExamWindowBean>forBoundedOutOfOrderness(Duration.ofSeconds(5L))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<DwsExaminationPaperExamWindowBean>() {
                                    @Override
                                    public long extractTimestamp(DwsExaminationPaperExamWindowBean dwsExaminationPaperExamWindowBean, long l) {
                                        return dwsExaminationPaperExamWindowBean.getTs();
                                    }
                                }
                        )
        );

        KeyedStream<DwsExaminationPaperExamWindowBean, String> keyedStream = withWaterMarkStream.keyBy(new KeySelector<DwsExaminationPaperExamWindowBean, String>() {
            @Override
            public String getKey(DwsExaminationPaperExamWindowBean value) throws Exception {
                return value.getPaperId();
            }
        });

        SingleOutputStreamOperator<DwsExaminationPaperExamWindowBean> reduce = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10))).reduce(
                new ReduceFunction<DwsExaminationPaperExamWindowBean>() {
                    @Override
                    public DwsExaminationPaperExamWindowBean reduce(DwsExaminationPaperExamWindowBean value1, DwsExaminationPaperExamWindowBean value2) throws Exception {
                        value1.setExamTakenCount(value1.getExamTakenCount() + value2.getExamTakenCount());
                        value1.setExamTotalDuringSec(value1.getExamTotalDuringSec() + value2.getExamTotalDuringSec());
                        value1.setExamTotalScore(value1.getExamTotalScore() + value2.getExamTotalScore());
                        return value1;
                    }
                },
                new ProcessWindowFunction<DwsExaminationPaperExamWindowBean, DwsExaminationPaperExamWindowBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<DwsExaminationPaperExamWindowBean, DwsExaminationPaperExamWindowBean, String, TimeWindow>.Context context, Iterable<DwsExaminationPaperExamWindowBean> elements, Collector<DwsExaminationPaperExamWindowBean> out) throws Exception {

                        String stt = DateFormatUtil.tsToDateTime(context.window().getStart());
                        String edt = DateFormatUtil.tsToDateTime(context.window().getEnd());
                        String curDate = DateFormatUtil.tsToDate(context.window().getStart());
                        for (DwsExaminationPaperExamWindowBean element : elements) {
                            element.setStt(stt);
                            element.setEdt(edt);
                            element.setCurDate(curDate);
                            element.setTs(System.currentTimeMillis());
                            out.collect(element);

                        }
                    }
                }
        );
        //reduce.print();
        SingleOutputStreamOperator<DwsExaminationPaperExamWindowBean> withProvinceDS = AsyncDataStream.unorderedWait(
                reduce,
                new DimAsyncFunction<DwsExaminationPaperExamWindowBean>() {
                    @Override
                    public void addDims(DwsExaminationPaperExamWindowBean obj, JSONObject dimJsonObj) {
                        obj.setPaperTitle(dimJsonObj.getString("paper_title"));
                        obj.setCourseId(dimJsonObj.getString("course_id"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_test_paper";
                    }

                    @Override
                    public String getRowKey(DwsExaminationPaperExamWindowBean obj) {
                        return obj.getPaperId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        SingleOutputStreamOperator<DwsExaminationPaperExamWindowBean> withexamds = AsyncDataStream.unorderedWait(
                withProvinceDS,
                new DimAsyncFunction<DwsExaminationPaperExamWindowBean>() {
                    @Override
                    public void addDims(DwsExaminationPaperExamWindowBean obj, JSONObject dimJsonObj) {
                        obj.setCourseName(dimJsonObj.getString("course_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_course_info";
                    }

                    @Override
                    public String getRowKey(DwsExaminationPaperExamWindowBean obj) {
                        return obj.getCourseId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        SingleOutputStreamOperator<String> map = withexamds.map(
                new MapFunction<DwsExaminationPaperExamWindowBean, String>() {
                    @Override
                    public String map(DwsExaminationPaperExamWindowBean bean) throws Exception {
                        SerializeConfig config = new SerializeConfig();
                        config.setPropertyNamingStrategy(PropertyNamingStrategy.SnakeCase);
                        String jsonStr = JSON.toJSONString(bean, config);
                        return jsonStr;
                    }
                }
        );
        map.print();

        map.sinkTo(FlinkSinkUtil.getDorisSink("dws_examination_paper_exam_window"));


    }
}
